package org.budo.warehouse.logic.consumer.jdbc;

import java.util.List;
import java.util.concurrent.ExecutorService;

import org.budo.support.lang.util.ListUtil;
import org.budo.support.lang.util.StringUtil;
import org.budo.warehouse.logic.api.DataEntry;

import lombok.extern.slf4j.Slf4j;

/**
 * @author lmw
 */
@Slf4j
public class MysqlDataConsumer extends JdbcDataConsumer {
    private static final String HA_HEALTH_CHECK = "ha_health_check";

    private static final String ALIBABA_RDS_ROW_ID = "#alibaba_rds_row_id#";

    @Override
    public ExecutorService getExecutorService() {
        return null; // 所有实例共用同一个线程池
    }

    /**
     * 追加 ON DUPLICATE KEY UPDATE 逻辑
     */
    @Override
    public SqlUnit dataRowInsertSql(DataEntry dataEntry, int rowIndex) {
        return this.dataRowInsertOnDuplicateKeyUpdate(dataEntry, rowIndex);
    }

    @Override
    public SqlUnit dataRowUpdateSql(DataEntry dataEntry, int rowIndex) {
        return this.dataRowInsertOnDuplicateKeyUpdate(dataEntry, rowIndex);
    }

    @Override
    public SqlUnit dataRowDeleteSql(DataEntry dataEntry, int rowIndex) {
        if (dataEntry.hasColumn(rowIndex, ALIBABA_RDS_ROW_ID)) {
            log.warn("#39 hasColumn " + ALIBABA_RDS_ROW_ID + ", skip this, dataEntry=" + dataEntry);
            return null;
        }

        return super.dataRowDeleteSql(dataEntry, rowIndex);
    }

    private SqlUnit dataRowInsertOnDuplicateKeyUpdate(DataEntry dataEntry, int rowIndex) {
        if (HA_HEALTH_CHECK.equals(dataEntry.getTableName())) {
            log.warn("#49 ha_health_check skip, dataEntry=" + dataEntry);
            return SqlUnit.NULL;
        }

        if (dataEntry.hasColumn(rowIndex, ALIBABA_RDS_ROW_ID)) {
            log.debug("#54 hasColumn {}, skip this, dataEntry={}", ALIBABA_RDS_ROW_ID, dataEntry);
            return SqlUnit.NULL;
        }

        SqlUnit dataRowInsertSql = super.dataRowInsertSql(dataEntry, rowIndex);
        List<Object> parameters = ListUtil.arrayToList(dataRowInsertSql.getParameters());

        List<String> dataRowUpdateSetSql = this.dataRowUpdateSetSql(dataEntry, rowIndex, parameters);
        String sql = dataRowInsertSql.getSql() + " ON DUPLICATE KEY UPDATE " + StringUtil.join(dataRowUpdateSetSql, ", ");

        return new SqlUnit(sql, parameters.toArray());
    }
}